package com.unittec.drainage.provider.log.service;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

import com.unittec.drainage.provider.log.Log;
import org.apache.commons.lang.StringUtils;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.unittec.drainage.provider.common.RabbitConstants;
import com.unittec.drainage.sdk.log.request.Req102011;
import com.unittec.drainage.sdk.log.request.Req102017;
import com.unittec.drainage.sdk.root.domain.GatewayRequest;
import com.unittec.drainage.sdk.root.domain.Result;

@Service
public class MQLogProducerService implements RabbitTemplate.ConfirmCallback {

    @Autowired
    private Log log;

    private final RabbitTemplate rabbitTemplate;

    Map<String, GatewayRequest<?>> reqMap = new ConcurrentHashMap<>(16);
    Map<String, Boolean> resMap = new ConcurrentHashMap<>(16);

    @Autowired
    public MQLogProducerService(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setConfirmCallback(this);
    }

    private CorrelationData logreq(GatewayRequest<?> req) {
        String id = StringUtils.isBlank(req.getRequestId()) ? UUID.randomUUID().toString() : req.getRequestId();
        CorrelationData correlationId = new CorrelationData(id);
        log.debug(req, "{} 消息投递", correlationId.toString());
        reqMap.put(id, req);
        return correlationId;
    }

    private boolean waitResponse(GatewayRequest<?> req, CorrelationData correlationId) {
        try {
            synchronized (req) {
                req.wait(1000);
            }
        } catch (InterruptedException e) {

        }
        boolean boo = resMap.get(correlationId.getId()) == null ? false : resMap.remove(correlationId.getId());

        if (!boo) {
            log.error(req, "消息发送失败");
            throw Result.exception("消息发送失败");
        }
        return boo;
    }

    public boolean createLog(GatewayRequest<Req102017> req) {
        req.setMethod("102017");
        CorrelationData correlationId = logreq(req);
        rabbitTemplate.convertAndSend(RabbitConstants.EXCHANGE_DIRECT, RabbitConstants.ROUTINGKEY_OPLOG,
            req.getContent(), correlationId);
        try {
            return waitResponse(req, correlationId);
        } catch (Exception e) {
            log.error(req, "操作日志消息发送失败", e);
            return false;
        }
    }

    public boolean sendLogLevelMsg(GatewayRequest<Req102011> req) {
        CorrelationData correlationId = logreq(req);
        rabbitTemplate.convertAndSend(RabbitConstants.EXCHANGE_FANOUT, null, req.getContent(), correlationId);
        return waitResponse(req, correlationId);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        GatewayRequest<?> req = reqMap.remove(correlationData.getId());
        if (null != req) {
            synchronized (req) {
                resMap.put(correlationData.getId(), ack);
                req.notifyAll();
            }
            if (!ack) {
                String logmsg = String.format("confirm requestId=[%s] appId=[%s], rid=[%s], cause=[%s]",
                    req.getRequestId(), req.getAppId(), correlationData.toString(), cause);
                log.sysErrLog("MQconfirm", logmsg, null);
            }
        }
    }
}
